package com.mall.hadoop.format;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

/**
 * 自定义LineRecordWriter类，继承抽象类RecordWriter
 * @author tgy
 *
 * @param <K> 需根据reducer的key类型传参
 * @param <V> 需根据reducer的value类型传参
 */
public class LineRecordWriter<K, V> extends RecordWriter<K, V> {  
	//输出格式为UTF-8
    private static final String utf8 = "UTF-8";  
    private static final byte[] newline;  
    static {  
        try { 
        	//换行符
        	newline = "\n".getBytes(utf8);
        } catch (UnsupportedEncodingException uee) {  
            throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
        }  
    }  
    protected DataOutputStream out;  
    //键值对划分符号
    private final byte[] keyValueSeparator;  
    //带两个参数的构造函数
    public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {  
        this.out = out;  
        try {  
            this.keyValueSeparator = keyValueSeparator.getBytes(utf8);  
        } catch (UnsupportedEncodingException uee) {  
            throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
        }  
    }  
    public LineRecordWriter(DataOutputStream out) { 
    	//调用上面带两个参数的构造方法
        this(out, "\\t");  
    }  
    private void writeObject(Object o) throws IOException {  
        if (o instanceof Text) {  
            Text to = (Text) o;  
            out.write(to.getBytes(), 0, to.getLength());  
        } else {  
            out.write(o.toString().getBytes(utf8));  
        }  
    }  
   //同步锁write方法，保证该方法在同一时间内有唯一线程执行，其它被阻塞
    public synchronized void write(K key, V value) throws IOException {  
        boolean nullKey = key == null || key instanceof NullWritable;  
        boolean nullValue = value == null || value instanceof NullWritable; 
        //如果key为null
        if (nullKey && nullValue) {  
            return;  
        }  
        if (!nullKey) {  
            writeObject(key);  
        }  
        if (!(nullKey || nullValue)) {  
            out.write(keyValueSeparator);  
        }  
        if (!nullValue) {  
            writeObject(value);  
        }  
        out.write(newline);  
    }  
    //同步锁close方法，保证该方法在同一时间内有唯一线程执行，其它被阻塞
    public synchronized void close(TaskAttemptContext context) throws IOException {  
        out.close();  
    }

}
